package org.apache.rocketmq.remoting.netty;

import io.netty.channel.ChannelDuplexHandler;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelPromise;
import org.apache.rocketmq.remoting.protocol.RemotingCommand;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.atomic.LongAdder;


/**
 * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
 * @author：陈清风扬，个人微信号：chenqingfengyangjj。
 * @date:2024/12/27
 * @方法描述：这个处理器是用来记录Namesrv接收请求和回复响应的频率的
 */
@ChannelHandler.Sharable
public class RemotingCodeDistributionHandler extends ChannelDuplexHandler {

    //存储接收到的各个请求的次数
    private final ConcurrentMap<Integer, LongAdder> inboundDistribution;

    //存储回复各个请求响应的次数
    private final ConcurrentMap<Integer, LongAdder> outboundDistribution;

    //构造方法
    public RemotingCodeDistributionHandler() {
        inboundDistribution = new ConcurrentHashMap<>();
        outboundDistribution = new ConcurrentHashMap<>();
    }

    //接收入站消息的方法
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        //判断消息类型是否为Remoting协议下的消息
        if (msg instanceof RemotingCommand) {
            //获得RemotingCommand消息
            RemotingCommand cmd = (RemotingCommand) msg;
            //统计该消息中接收到的请求的次数
            countInbound(cmd.getCode());
        }
        ctx.fireChannelRead(msg);
    }


    //统计入站消息的次数
    private void countInbound(int requestCode) {
        //判断map中是否存储了对应的请求码的信息，如果没有则创建一个
        LongAdder item = inboundDistribution.computeIfAbsent(requestCode, k -> new LongAdder());
        //在这里对该请求码的次数自增
        item.increment();
    }

    //发送消息出站的方法
    @Override
    public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
        //判断消息类型是否为Remoting协议下的消息
        if (msg instanceof RemotingCommand) {
            //获得RemotingCommand消息
            RemotingCommand cmd = (RemotingCommand) msg;
            //统计要回复的响应的次数
            countOutbound(cmd.getCode());
        }
        ctx.write(msg, promise);
    }

    //统计出站消息次数的方法
    private void countOutbound(int responseCode) {
        LongAdder item = outboundDistribution.computeIfAbsent(responseCode, k -> new LongAdder());
        item.increment();
    }


    //得到入站消息次数的字符串快照
    public String getInBoundSnapshotString() {
        return this.snapshotToString(this.getDistributionSnapshot(this.inboundDistribution));
    }


    //得到出站消息次数的字符串快照
    public String getOutBoundSnapshotString() {
        return this.snapshotToString(this.getDistributionSnapshot(this.outboundDistribution));
    }



    //该方法会把一个map中的内容复制到一个新的map中，然后返回这个新的map
    private Map<Integer, Long> getDistributionSnapshot(Map<Integer, LongAdder> countMap) {
        Map<Integer, Long> map = new HashMap<>(countMap.size());
        for (Map.Entry<Integer, LongAdder> entry : countMap.entrySet()) {
            map.put(entry.getKey(), entry.getValue().sumThenReset());
        }
        return map;
    }


    //该方法会把一个map中的内容拼接成字符串，然后把字符串返回
    private String snapshotToString(Map<Integer, Long> distribution) {

        if (null != distribution && !distribution.isEmpty()) {
            StringBuilder sb = new StringBuilder("{");
            boolean first = true;
            for (Map.Entry<Integer, Long> entry : distribution.entrySet()) {
                if (0L == entry.getValue()) {
                    continue;
                }
                sb.append(first ? "" : ", ").append(entry.getKey()).append(":").append(entry.getValue());
                first = false;
            }
            if (first) {
                return null;
            }
            sb.append("}");
            return sb.toString();
        }
        return null;
    }
}
